package org.example.flink.eventtime;

import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.OperatorStateStore;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.runtime.ValueSerializer;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.checkpoint.Checkpoint;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.table.expressions.In;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @author zhangjg
 * @desc 自定义Source
 * @date 2022/01/08 10:11
 **/
public class CustomSource extends RichSourceFunction<S0> implements CheckpointedFunction {

    private static final Logger log = LoggerFactory.getLogger(CustomSource.class);
    private volatile boolean running = true;
    private transient volatile int num = 1;
    private transient ListState<Integer> listState;

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        listState.clear();
        listState.add(num);
        log.info("snapshotState......num={}",num);
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        OperatorStateStore stateStore = context.getOperatorStateStore();
        ListStateDescriptor listStateDescriptor = new ListStateDescriptor("checkPointedList",
                TypeInformation.of(new TypeHint<Integer>() {}));
        listState = context.getOperatorStateStore().getListState(listStateDescriptor);
        if(context.isRestored()){
            num = listState.get().iterator().next();
        }
        log.info("initializeState......num={}",num);
    }

    @Override
    public void run(SourceContext<S0> ctx) throws Exception {
        String[] arr = {"a","b","c"};
        Random random = new Random();
        long time = DateUtil.date().getTime();
        while (running){
            S0 s = new S0();
            s.setId(num);
            s.setName(arr[random.nextInt(2)]);
            s.setEventtime(DateUtil.date(time).toString("yyyy-MM-dd HH:mm:ss"));
            ctx.collect(s);
            num++;
            time += 1*1000;
//            log.info(JSONObject.toJSONString(s));
            Thread.sleep(TimeUnit.SECONDS.toMillis(1));
        }
    }

    @Override
    public void cancel() {
        running = false;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }

    @Override
    public void close() throws Exception {
        super.close();
    }
}
